/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package org.just.dusin.ooii.component.standard;

import java.util.Properties;


import java.util.logging.Level;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;

import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.RedeliveryPolicy;
import org.apache.log4j.Logger;
import org.just.dusin.ooii.component.Source;
import org.dusin.webcoal.exception.OOIIException;

/**
 *
 * @author Administrator
 */
public class AMQSource extends Source {

    private String brokerName="broke";
    private static final Logger logger= Logger.getLogger(AMQSource.class.getName());
    public String getBrokerName() {
        return brokerName;
    }

    public void setBrokerName(String brokerName) {
        this.brokerName = brokerName;
    }

    public String getBrokerURL() {
        return brokerURL;
    }

    public void setBrokerURL(String brokerURL) {
        this.brokerURL = brokerURL;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public String getQueueName() {
        return queueName;
    }

    public void setQueueName(String queueName) {
        this.queueName = queueName;
    }

    public String getUser() {
        return user;
    }

    public void setUser(String user) {
        this.user = user;
    }
    private String brokerURL = "tcp://localhost:61616";
    private String queueName="coal";    
    private int inteval=300;

    public void setInteval(int inteval) {
        this.inteval = inteval;
    }
    
    private String user;
    private String password;     
    @Override
    public void initComponent(Properties props) {
    }

    @Override
    public void destroyComponent() throws Exception {
    }

    @Override
    public void run() {
       
        while (!stop) {
            ConnectionFactory factory = new ActiveMQConnectionFactory(brokerURL);
            int i=0;
        
            Connection connection =null;
            Session session = null;
            MessageConsumer messageConsumer =null;
            try {
                
                connection = factory.createConnection(user, password);
                RedeliveryPolicy policy = ((ActiveMQConnection)connection).getRedeliveryPolicy();
                policy.setRedeliveryDelay(100);
                policy.setMaximumRedeliveries(-1);
                connection.start();
                //session.start();
                session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
                Destination destination = session.createQueue(queueName);
                messageConsumer = session.createConsumer(destination);
                TextMessage txtMessage = (TextMessage)messageConsumer.receive(1000);                
                if(txtMessage!=null){
                    String msgText = txtMessage.getText();
                    logger.info("处理消息"+msgText);
                    sink.process(txtMessage.getText());
                    session.commit();
                }else{
                    logger.debug("无消息");
                }
            
               
                Thread.sleep(inteval);
            } catch (JMSException e) {
                if(session!=null){
                    try {
                        session.rollback();
                    } catch (JMSException ex) {
                        logger.error("事物回滚异常");
                    }
                }
                logger.error(e.getMessage());
            }catch(OOIIException e){
                try {
                    session.rollback();
                } catch (JMSException ex) {
                    logger.error("事物回滚异常");
                }
                logger.error(e.getMessage());
            }catch(InterruptedException e){
                logger.error(e.getMessage());
            }catch(Exception e){
                logger.error(e.getMessage());
            }finally{
                try { 
                    try{
                        if (messageConsumer != null) {
                            messageConsumer.close();
                        }
                        if (session != null) {
                            session.close();
                        }
                        if (connection != null) {
                            connection.close();
                        }
                    }catch(JMSException e){
                        logger.error("jsm 关闭错误");
                    }
                    
                    Thread.sleep(inteval);
                } catch (InterruptedException ex) {
                    logger.warn("睡眠中断");
                }
            }            
        }
    }
}
